1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package rx.internal.operators;
17
18 import rx.Observable.Operator;
19 import rx.Subscriber;
20 import rx.functions.Func1;
21 import rx.functions.Func2;
22
23
24
25
26
27 public final class OperatorSkipWhile<T> implements Operator<T, T> {
28 private final Func2<? super T, Integer, Boolean> predicate;
29
30 public OperatorSkipWhile(Func2<? super T, Integer, Boolean> predicate) {
31 this.predicate = predicate;
32 }
33 @Override
34 public Subscriber<? super T> call(final Subscriber<? super T> child) {
35 return new Subscriber<T>(child) {
36 boolean skipping = true;
37 int index;
38 @Override
39 public void onNext(T t) {
40 if (!skipping) {
41 child.onNext(t);
42 } else {
43 if (!predicate.call(t, index++)) {
44 skipping = false;
45 child.onNext(t);
46 } else {
47 request(1);
48 }
49 }
50 }
51
52 @Override
53 public void onError(Throwable e) {
54 child.onError(e);
55 }
56
57 @Override
58 public void onCompleted() {
59 child.onCompleted();
60 }
61 };
62 }
63
64 public static <T> Func2<T, Integer, Boolean> toPredicate2(final Func1<? super T, Boolean> predicate) {
65 return new Func2<T, Integer, Boolean>() {
66
67 @Override
68 public Boolean call(T t1, Integer t2) {
69 return predicate.call(t1);
70 }
71 };
72 }
73 }